package kassiopeia.agent.seda.threads;

import kassiopeia.agent.seda.stage.Stage;
import org.apache.log4j.Logger;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class SedaPool {

    private Logger log = Logger.getLogger(SedaPool.class);

    private ExecutorService threadPool;
    private List<Stage> stages;

    public SedaPool(List<Stage> stages) {
        this.stages = stages;
        threadPool = Executors.newCachedThreadPool();
    }

    public void start() {
        log.info("Starting stages...");
        for (Stage stage : stages) {
            threadPool.execute(stage);
        }
    }

    public void shutdown() {
        log.info("Shutting down SEDA pool...");
        for (Stage stage : stages) {
            stage.stop();
        }
        threadPool.shutdown();
        waitForStagesToTerminate();
        log.info("SEDA pool shutdown");
    }

    private void waitForStagesToTerminate() {
        log.debug("SEDA pool terminating...");
        while (!threadPool.isTerminated()) {
            try {
                TimeUnit.SECONDS.sleep(1L);
            } catch (InterruptedException ex) {
                log.error(ex, ex);
            }
        }
        log.info("SEDA pool terminated");
    }
}
